package lucene.server.cdc;

import lucene.server.cdc.client.AbstractClient;
import lucene.server.cdc.mysql.MysqlBinlogClient;
import lucene.server.cdc.oracle.OracleLogMinerClient;
import lucene.server.cdc.postgresql.PostgresqlClient;
import lucene.server.cdc.sqlserver.SqlServerClient;
import lucene.server.common.task.ITask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.Date;
import java.util.Properties;
import java.util.concurrent.ScheduledFuture;

public class CDCTask  implements Runnable , ITask {
    private static final Logger log = LoggerFactory.getLogger(CDCTask.class);
    Properties properties;
    AbstractClient client;
    private ScheduledFuture future;
    public CDCTask(Properties properties)
    {
        this.properties = properties;
        client = createClient(properties);
        client.start();
    }
    @Override
    public void run() {
        if(!client.isRunning()){
            client.reStart();
        }
    }

    @Override
    public String getName() {
        return "cdc_"+properties.getProperty("name");
    }

    @Override
    public void shutdown() {
        client.shutdown();
        if (this.future != null) {
            this.future.cancel(true);
        }
    }

    @Override
    public boolean isRunning() {
        return client.isRunning();
    }

    @Override
    public ScheduledFuture createScheduled(ThreadPoolTaskScheduler threadPoolTaskScheduler) {
        Date startTime = new Date( System.currentTimeMillis()+10000);
        this.future = threadPoolTaskScheduler.scheduleWithFixedDelay(this,startTime,10000);
        return future;
    }

    @Override
    public String getTaskPlan() {
        return "触发运行";
    }

    public static AbstractClient createClient(Properties properties){
        String databaseType = properties.getProperty("type").toLowerCase();
        switch (databaseType){
            case "mysql":
                return new MysqlBinlogClient(properties);
            case "oracle":
                return new OracleLogMinerClient(properties);
            case "postgresql":
                return new PostgresqlClient(properties);
            case "sqlserver":
                return new SqlServerClient(properties);
            default:
                return null;
        }
    }
}
